package org.geeks.gsjetcache.listener;

import org.geeks.gsjetcache.constant.JetcachePurge;
import org.geeks.gsjetcache.service.JetcachePurgeService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import io.lettuce.core.RedisClient;
import io.lettuce.core.pubsub.RedisPubSubListener;
import io.lettuce.core.pubsub.StatefulRedisPubSubConnection;
import io.lettuce.core.pubsub.api.async.RedisPubSubAsyncCommands;
import lombok.extern.slf4j.Slf4j;

@Slf4j
@Component
public class JetcacheConsumer {

	@Autowired
	private JetcachePurgeService jetcachePurgeService;

	public JetcacheConsumer(RedisClient defaultClient) {
		StatefulRedisPubSubConnection<String, String> connection = defaultClient.connectPubSub();

		connection.addListener(new RedisPubSubListener<String, String>() {

			@Override
			public void unsubscribed(String channel, long count) {
				// TODO Auto-generated method stub

			}

			@Override
			public void subscribed(String channel, long count) {
				// TODO Auto-generated method stub

			}

			@Override
			public void punsubscribed(String pattern, long count) {
				// TODO Auto-generated method stub

			}

			@Override
			public void psubscribed(String pattern, long count) {
				// TODO Auto-generated method stub

			}

			@Override
			public void message(String pattern, String channel, String message) {
				// TODO Auto-generated method stub

			}

			@Override
			public void message(String channel, String message) {
				log.info("Receive message {} : {}", channel, message);

				String[] infos = message.split(":");

				if ("byId".equals(infos[0])) {
					jetcachePurgeService.byId(Long.parseLong(infos[1]));
				}
			}
		});

		RedisPubSubAsyncCommands<String, String> async = connection.async();

		async.subscribe(JetcachePurge.CHANNEL);
	}
}
